package com.zhang.gmall.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.beans.PageViewStats;
import com.zhang.gmall.utils.ClickHouseUtil;
import com.zhang.gmall.utils.DateTimeUtil;
import com.zhang.gmall.utils.KafkaUtil;
import com.zhang.gmall.utils.MyEnv;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;

/**
 * @title: pv 统计分析
 * @author: zhang
 * @date: 2022/3/27 15:43
 */
public class PageViewStats10sApp {
    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = MyEnv.getStreamingEnv(4, false);

        //TODO 2.读取kafka数据，转换POJO类,分配水位线和提取时间字段，按维度分组聚合
        String sourceTopic = "dwd_page_log";
        String groupId = "page_view_stats_10s_app";
        SingleOutputStreamOperator<PageViewStats> pvStatsDS = env
                .addSource(KafkaUtil.getKafkaSource(sourceTopic, groupId))
                .map(new MapFunction<String, PageViewStats>() {
                    @Override
                    public PageViewStats map(String value) throws Exception {
                        JSONObject logJson = JSON.parseObject(value);
                        JSONObject common = logJson.getJSONObject("common");
                        JSONObject pageLog = logJson.getJSONObject("page");
                        String lastPageId = pageLog.getString("last_page_id");
                        long sv_ct = 0L;
                        if (lastPageId == null || lastPageId.length() <= 0) {
                            sv_ct = 1L;
                        }
                        return new PageViewStats(
                                "", "",
                                pageLog.getString("page_id"),
                                common.getString("vc"),
                                common.getString("ch"),
                                common.getString("ar"),
                                common.getString("is_new"),
                                1L, sv_ct,
                                pageLog.getLong("during_time"),
                                logJson.getLong("ts")
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<PageViewStats>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<PageViewStats>() {
                                    @Override
                                    public long extractTimestamp(PageViewStats element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                })
                )
                .keyBy(new KeySelector<PageViewStats, Tuple5<String, String, String, String, String>>() {
                    @Override
                    public Tuple5<String, String, String, String, String> getKey(PageViewStats value) throws Exception {
                        return Tuple5.of(
                                value.getPage_id(),
                                value.getAr(),
                                value.getCh(),
                                value.getVc(),
                                value.getIs_new()
                        );
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<PageViewStats>() {
                    @Override
                    public PageViewStats reduce(PageViewStats value1, PageViewStats value2) throws Exception {
                        value1.setDur_time(value1.getDur_time() + value2.getDur_time());
                        value1.setPv_ct(value1.getPv_ct() + value2.getPv_ct());
                        value1.setSv_ct(value1.getSv_ct() + value2.getSv_ct());
                        return value1;
                    }
                }, new WindowFunction<PageViewStats, PageViewStats, Tuple5<String, String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple5<String, String, String, String, String> stringStringStringStringStringTuple5, TimeWindow window, Iterable<PageViewStats> input, Collector<PageViewStats> out) throws Exception {
                        PageViewStats pageViewStats = input.iterator().next();
                        //补充窗口信息
                        pageViewStats.setStt(DateTimeUtil.toYMDhms(new Date(window.getStart())));
                        pageViewStats.setEdt(DateTimeUtil.toYMDhms(new Date(window.getEnd())));
                        //补充处理时间
                        pageViewStats.setTs(System.currentTimeMillis());
                        //向下传递数据数据
                        out.collect(pageViewStats);
                    }
                });
        //TODO 3.将数据写入clickhouse
        pvStatsDS.addSink(
                ClickHouseUtil.getClickHouseSink(
                        "insert into dws_pv_vc_ch_isnew_ar_10s values(?,?,?,?,?,?,?,?,?,?,?)"
                )
        );

        //打印测试
        pvStatsDS.print(">>>>>");
        //TODO 4.执行任务
        env.execute();
    }
}
